package com.lock.common;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 基于临时有序节点的分布式锁
 * 这里采用的是递归的方式，而不是监听的方式
 * Created by Administrator on 2017/7/14.
 */
public class FBSOrderLock {
    public final static int num = 5;
    public static CountDownLatch countDownLatch = new CountDownLatch(FBSOrderLock.num);

    public static void main(String[] args) throws InterruptedException {
        for (int i = 1;i<=num;i++){
            OrderLock orderLock = new OrderLock();
            new Thread(orderLock).start();
        }
        countDownLatch.await();
        System.out.println("所有线程已全部执行完");
    }
}
class OrderLock implements Watcher,Runnable{
    private ZooKeeper zooKeeper;
    private String nowPath;
    private String prePath;
    private String name;
    private final static String parentPath = "/com.lock.common.OrderLock";
    private final static String path = "/com.lock.common.OrderLock/sub";

    public OrderLock() {
        try {
            zooKeeper = new ZooKeeper("127.0.0.1:2181",60000,this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void process(WatchedEvent watchedEvent) {
        if ( Event.KeeperState.SyncConnected == watchedEvent.getState()) {
            if ( Event.EventType.NodeCreated == watchedEvent.getType() ) {
                System.out.println(Thread.currentThread().getName()+"成功连接上ZK服务器");
            } else if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(prePath)) {
                System.out.println(name+"收到情报，排我前面的家伙已挂，我是不是可以出山了？");
                if(isMin()){
                    workIng();
                }
            }
        }
    }

    public void run() {
        name = Thread.currentThread().getName();
        caeateParentPath();
        //获取锁
        if (getLock()) {
            //操作
            workIng();
        }
    }

    /**
     * 工作
     */
    private void workIng(){
        try {
            //exists表示监听指定节点的变化情况，会回调监听方法
            if(zooKeeper.exists(this.nowPath,true) == null){
                System.out.println(name+"本节点已不在了...");
                return;
            }
            System.out.println(name+"获取锁成功，赶紧干活！===========");
            zooKeeper.delete(nowPath,-1);
            System.out.println(name+"删除本节点："+nowPath);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            FBSOrderLock.countDownLatch.countDown();
            releaseConnection();
        }
    }

    /**
     * 获取锁
     * @return
     */
    private Boolean getLock() {
        nowPath = createPath(path);
        System.out.println(name+"=======创建子节点"+nowPath);
        //判断自己是不是最小节点
        return isMin();
    }

    /**
     * 当前线程节点是否是最小的
     * @return
     */
    private boolean isMin() {
        try {
            List<String> childList = zooKeeper.getChildren(parentPath,false);
            Collections.sort(childList);
            int index = childList.indexOf(nowPath.substring(parentPath.length()+1));
            switch (index){
                case -1:{
                    System.out.println(name+"=该线程节点"+path+"不存在");
                    return false;
                }
                case 0:{
                    System.out.println(name+"=该线程是老大，允许执行");
                    return true;
                }
                default:{
                    prePath = parentPath+"/"+childList.get(index-1);
                    System.out.println(name+"=获取子节点中，排在我面前的是"+prePath);
                    //exists表示监听前一个节点的变化情况，会回调监听方法
                    zooKeeper.exists(prePath,true);
                    return false;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
    }

    private void caeateParentPath(){
        synchronized (this){
            if (createPath(parentPath,"111")){
                System.out.println(name+"=线程抢占先机创建了主节点。。。。");
            }
        }
    }

    private Boolean createPath(String path,String data){
        try {
            zooKeeper.create(path,data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            return true;
        } catch (KeeperException e) {
            return false;
        } catch (InterruptedException e) {
            return false;
        }
    }

    private String createPath(String path){
        try {
            return zooKeeper.create(path,"111".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (KeeperException e) {
            e.printStackTrace();
            return "";
        } catch (InterruptedException e) {
            e.printStackTrace();
            return "";
        }
    }

    /**
     * 关闭ZK连接
     */
    private void releaseConnection() {
        if ( this.zooKeeper !=null ) {
            try {
                this.zooKeeper.close();
            } catch ( InterruptedException e ) {}
        }
        System.out.println(name+"释放连接");
    }
}
